package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.apache.catalina.User;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService {

    @Autowired
    TagInfoService tagInfoService;

    /*
            将userGroup处理:
                    ①把List<TagConditoin> 转为 jsonstr
                    ②为create_time赋值当前时间
                    ③生成注释:  conditionComment=null
                            队友提供了现成的方法:   UserGroup.conditionJsonToComment
     */
    @Override
    public void saveUserGroupToMysql(UserGroup userGroup) {

        String jsonStr = JSON.toJSONString(userGroup.getTagConditions());
        userGroup.setConditionJsonStr(jsonStr);

        String comment = userGroup.conditionJsonToComment();
        userGroup.setConditionComment(comment);

        userGroup.setCreateTime(new Date());

        //存入Mysql
        saveOrUpdate(userGroup);

    }



    //计算人群包的人数
    @Override
    public UserGroup countUserGroupNum(UserGroup userGroup) {

        //生成bitmapAndSql
        String sql = combineBitmapAnd(userGroup);

        //写入到ck
        super.baseMapper.saveUserGroupToCk(userGroup.getId().toString(),sql);

        //从ck中查询人群包，写入redis
        saveUserGroupToRedis(userGroup);

        //查询人数
        Long num = super.baseMapper.queryUserGroupNum(sql);

        userGroup.setUserGroupNum(num);

        return userGroup;

    }

    // 既能写入也能更新
    public Long saveUserGroupToRedis(UserGroup userGroup) {

        //Jedis(redis的java客户端)
        Jedis jedis = RedisUtil.getJedis();

        //从ck中查询到了用户分群信息
        // 在更新功能中，废弃。 CK的更新是异步的。执行了update语句后，数据只是打上了删除标记，会在合并时，真正删除。
        // 一定在合并之前，查询了ck，意味着删除的数据和新更新的数据都会查到
        //List<String> uids = super.baseMapper.queryUserGroupFromCkById(userGroup.getId().toString());
        String sql = combineBitmapAnd(userGroup);

        List<String> uids = super.baseMapper.queryUserGroupFromCkById2(sql);

        //自己去计算指定人群的人群包


        /*
            K:  字符串，唯一性
                    id作为key
            V:  单值: string,hash
                        不方便
                集合:  list,set,zset
                        选择set， sismember()可以判断一个用户是否在集合中。画像常用的。

                  场景： 人群包，男性00后:  [1,2,3...]
                        set.sismember(5)

                   选择set
         */
        String[] set = new String[]{};
        set = uids.toArray(set);

        String key = "usergroup_" + userGroup.getId();
        //先删除key，防止有旧数据存在
        jedis.del(key);
        jedis.sadd(key,set);

        //把连接还入池子
        jedis.close();

        return uids.size() + 0l;

    }

    @Override
    public Long evaluateNum(UserGroup userGroup) {

        String sql = combineBitmapAnd(userGroup);

        return super.baseMapper.queryUserGroupNum(sql);
    }

    @Override
    public void updateUserGroupInCK(UserGroup userGroup) {

        String id = userGroup.getId().toString();
        //先删除旧的信息
        super.baseMapper.deleteUserGroup(id);

        //再写入新的信息
        String sql = combineBitmapAnd(userGroup);
        super.baseMapper.saveUserGroupToCk(id,sql);

    }

    /*
    bitmapAnd(
            bitmapAnd( (sql1),(sql2) ) ,
            (sql3)
          )
          .....

          数量关系:  bitmapAnd的数量 =  sql的数量 - 1
          只有1个sql时，无需拼接bitmapAnd。从第二个sql开始，每出现一个sql，拼接一个bitmapAnd


           bitmapAnd( bitmapAnd( (sql1),(sql2)) ,(sql3))

     */
    public String combineBitmapAnd(UserGroup userGroup){

        //获取前台传入的所有条件
        List<TagCondition> tagConditions = userGroup.getTagConditions();

        //查询到所有的三级标签信息
        Map<String, TagInfo> codeMap = tagInfoService.getTagInfoMapWithCode();

        //查询的业务日期
        String doDate = userGroup.getBusiDate();

        //准备生成最终sql的字符串追加器
        StringBuilder builder = new StringBuilder();

        //为每一天TagCondition生成子查询的sql
        for (TagCondition tagCondition : tagConditions) {

            String conditionSql = generateSingleTagConditionStr(tagCondition, codeMap, doDate);

            if (builder.length() == 0){
                builder.append("(").append(conditionSql).append(")");
            }else{
                //不是第一个条件
                builder.insert(0," bitmapAnd( ")
                       .append(",(")
                       .append(conditionSql)
                       .append("))");
            }

        }

        String sql = builder.toString();

        return sql;

    }


    /*
            根据用户传入的一个条件，生成一个子查询

         select
                groupBitmapMergeState(us)
         from ①
         where tag_code = '②'
         and tag_value ③ ('④', '女性')
         and dt='⑤'

            分析:
                    ①user_tag_value_string 来源表名，是变量。根据查询的tag的值的类型，判断从ck的4张bitmap表的哪个表查询
                    ②查询的是哪个标签
                            来自前台传入的条件    TagCondition.tagCode
                    ③条件的组合方式(运算符)
                                映射operator": "eq" 为实际的运算符，例如 =
                                如果运算符是范围查询(in,not in),条件需要添加 ()

                     ④传入的查询条件
                                一共四种类型的标签：
                                        decail,long:  查询条件值无需处理
                                        string,date:  查询条件需要手动拼接单引号

                      ⑤业务日期

               TagCondition tagCondition： 前台封装的查询条件，没有tagValueType(单独去查询)
                                队友写好了 TagInfoServiceImpl.getTagInfoMapWithCode()： 获取到所有定义的三级标签
                                        返回 Map<String,TagInfo>
                                                    key:  标签的tagCode
                                                    value: TagInfo

     */
    private String generateSingleTagConditionStr(TagCondition tagCondition,Map<String,TagInfo> tagInfoMap,String doDate){

        String template = "  select  groupBitmapMergeState(us) from %s where tag_code = '%s' and tag_value %s %s and dt='%s'";

        //必须获取每一个标签，所计算的值的类型(tagValueType) TagInfo
        String tagCode = tagCondition.getTagCode();
        TagInfo tagInfo = tagInfoMap.get(tagCode);
        String tagValueType = tagInfo.getTagValueType();

        //对tagValueType进行判断，确定这个值在ck的哪个bitmap表中
        String sourceTable = "";

        switch (tagValueType){
            case ConstCodes.TAG_VALUE_TYPE_LONG : sourceTable = "user_tag_value_long"; break;
            case ConstCodes.TAG_VALUE_TYPE_DECIMAL : sourceTable = "user_tag_value_decimal"; break;
            case ConstCodes.TAG_VALUE_TYPE_STRING : sourceTable = "user_tag_value_string"; break;
            case ConstCodes.TAG_VALUE_TYPE_DATE : sourceTable = "user_tag_value_date"; break;
        }

        //获取操作符
        String operator = tagCondition.getOperator();
        String operatorStr = getConditionOperator(operator);

        // [ 男,女,未知]   拼接完   '男','女','未知'
        List<String> tagValues = tagCondition.getTagValues();
        //封装传入的最终的条件
        String conditionStr = "";
        if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagValueType) || ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagValueType)){

            //传入的条件中的每个元素都需要拼接 '
            conditionStr = "'" +StringUtils.join(tagValues, "','") + "'";

        }else{
            // [200,300]  拼接完  200,300
            conditionStr = StringUtils.join(tagValues, ",");
        }

        //判断是否是in 或 not in，如果是在条件的外层添加 ()
        if ("in".equals(operator) || "nin".equals(operator)){
           conditionStr =  "(" + conditionStr +")";
        }

        //格式化
        String sql = String.format(template, sourceTable, tagCode.toLowerCase(), operatorStr, conditionStr, doDate);

        System.out.println(sql);

        return sql;

    }

    private String getConditionOperator(String operator){

        switch (operator) {
            case "eq" : return "=";
            case "neq" : return "<>";
            case "gte" : return ">=";
            case "lte" : return "<=";
            case "gt" : return ">";
            case "lt" : return "<";
            case "in" : return "in";
            case "nin" : return "not in";
        }
        throw new RuntimeException("没有匹配的操作符!");

    }
}
